package com.heima.datastructure.blockingqueue;

import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 单锁实现阻塞队列
 *
 * @param <E> 队列元素的存储类型
 * @author 勾新杰
 * @version 1.0
 * @date 2024/9/4 15:28
 */
public class BlockingQueue1<E> implements BlockingQueue<E>, Iterable<E> {
    private final E[] arr; // 数组
    private int head; // 头指针，指向第一个元素
    private int tail; // 尾指针，指向最后一个元素的下一个元素
    private int size; // 当前队列的大小

    // 1、ReentrantLock的可重入是指一个线程可以多次获得该锁，允许在有锁的情况下再次请求锁而不会发生死锁
    // 2、ReentrantLock可以指定公平锁还是非公平锁，公平锁就是按照顺序来获取锁，非公平锁相反，但是后者的性能更好
    // 3、ReentrantLock可以尝试非阻塞的获取锁，即如果获取到的锁不可用就直接返回false，而不是阻塞等待，避免不必要的等待
    // 4、ReentrantLock可中断的锁获取，即线程在等待锁时可以响应中断请求，避免一直被阻塞，synchronized就没有这个功能
    // 5、ReentrantLock支持条件变量，提供一个或多个Condition对象，用于线程间的协调，Condition对象提供了非常灵活的条件等待和通知机制
    private final ReentrantLock lock = new ReentrantLock(); // 可重入锁

    // 1、Condition是与ReentrantLock紧密相关的一个类，用于实现线程间的协调，一个ReentrantLock可以有多个条件变量
    // 2、Condition可以实现指定条件下线程的唤醒和等待
    // 3、根据条件阻塞的线程可以在Condition上等待，直到其他线程调用该Condition的signal方法或者signalAll来通知唤醒它
    // 4、它继承了 ReentrantLock 的所有特性，包括可重入性、公平性、尝试非阻塞地获取锁以及可中断的锁获取等。
    // 5、await和signal必须是在同步代码块内调用
    private final Condition emptyWait = lock.newCondition(); // 条件变量：每个因为链表为空导致阻塞的poll线程都加入到emptyWait中等待其他线程调用emptyWait的signal方法来唤醒它
    private final Condition fullWait = lock.newCondition(); // 条件变量：每个因为链表为满导致阻塞的offer线程都加入到fullWait中等待其他线程调用fullWait的signal方法来唤醒它

    /**
     * 用构造方法实现初始化
     *
     * @param capacity 容量
     */
    @SuppressWarnings("all")
    public BlockingQueue1(int capacity) {
        arr = (E[]) new Object[capacity];
    }

    /**
     * 元素入队
     *
     * @param value 入队元素
     * @throws InterruptedException 中断异常
     */
    @Override
    public void offer(E value) throws InterruptedException {
        lock.lockInterruptibly(); // 可响应中断请求的锁，保证不会被一直阻塞
        try {
            while (isFull()) { // 防止虚假唤醒，就是唤醒了不是第一个抢到锁的，可能有线程已经调用了offer方法导致队列为满
                fullWait.await(); // 阻塞当前线程，并释放锁，确保当前线程等待期间不持有锁，当此线程被poll线程唤醒的时候，就会先去抢锁，然后再从这行代码继续往下执行，就进入了while循环
            }
            arr[tail] = value;
            tail = inc(tail);
            size++;
            // 有元素入队后就说明队列中有一个元素了，也就是说生产者线程offer生产了一个元素，那么就该唤醒一个消费者线程poll
            emptyWait.signal(); // 唤醒一个阻塞线程，不释放锁
        } finally {
            lock.unlock(); // 解锁，有异常的时候也要解锁
        }
    }

    /**
     * 阻塞队列入队
     *
     * @param value   入队元素
     * @param timeout 当前线程阻塞后等待时间上限，限制单位是毫秒
     * @return 当前线程阻塞等待总时间超过timeout，或者说timout毫秒之后队列还是满，则返回false返回false，其余肯定入队成功，返回true
     * @throws InterruptedException 中断异常
     */
    @Override
    public boolean offer(E value, long timeout) throws InterruptedException {
        lock.lockInterruptibly(); // 可响应中断请求的锁，保证不会被一直阻塞
        try {
            long remain = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isFull()) { // 防止虚假唤醒，就是唤醒了不是第一个抢到锁的，可能有线程已经调用了offer方法导致队列为满
                if (remain <= 0) { // 已经判断了队列是满并且剩余等待时间已经无了，再等待没意义，返回false
                    return false;
                }
                // 等待时间未被耗尽时，如果被其他线程唤醒，而且没有抢到锁，而且其它生产者线程offer把队列装满了，那么此生产者线程又会被阻塞，所以要保证阻塞的总时间不超过timeout，而不是单次等待
                remain = fullWait.awaitNanos(remain);// 阻塞当前线程，并限制等待时间，返回的是剩余的等待时间，并释放锁，确保当前线程等待期间不持有锁，当此线程被poll线程唤醒的时候，就会先去抢锁，然后再从这行代码继续往下执行，就进入了while循环
            }
            arr[tail] = value;
            tail = inc(tail);
            size++;
            // 有元素入队后就说明队列中有一个元素了，也就是说生产者线程offer生产了一个元素，那么就该唤醒一个消费者线程poll
            emptyWait.signal();
            return true;
        } finally {
            lock.unlock(); // 解锁，有异常的时候也要解锁
        }
    }

    /**
     * 获取队尾元素并删除
     *
     * @return 队尾元素
     * @throws InterruptedException 中断异常
     */
    @Override
    public E poll() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (isEmpty()) { // 防止假唤醒
                emptyWait.await(); // 阻塞当前线程，并释放该线程持有的锁，将当前线程加入到emptyWait等待，当被生产者线程offer唤醒的时候，应先抢到锁，再从当前代码继续往下执行，进行while循环
            }
            E value = arr[head];
            arr[head] = null; // 垃圾回收
            head = inc(head);
            size--;
            // 当消费者线程poll消费了一个元素，就应该唤醒一个生产者线程offer
            fullWait.signal();
            return value;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 出队
     *
     * @param timeout 时间上限
     * @return 被阻塞timeout时间后或者说timeout毫秒时间后如果队列还是空，就返回null，否则返回出队元素
     * @throws InterruptedException 中断异常
     */
    @Override
    public E poll(long timeout) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            long remain = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isEmpty()) {
                if (remain <= 0) {
                    return null;
                }
                remain = emptyWait.awaitNanos(remain);
            }
            E value = arr[head];
            arr[head] = null;
            head = inc(head);
            size--;
            fullWait.signal();
            return value;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 判断是否为空
     *
     * @return 为空返回true
     */
    public boolean isEmpty() {
        return size == 0;
    }

    /**
     * 判断是否为满
     *
     * @return 为满返回true
     */
    public boolean isFull() {
        return size == arr.length;
    }

    /**
     * 索引自增操作
     *
     * @param index 索引
     * @return 自增后的结果
     */
    private int inc(int index) {
        return ++index == arr.length ? 0 : index;
    }

    /**
     * 索引自减操作
     *
     * @param index 索引
     * @return 自减后的结果
     */
    private int dec(int index) {
        return --index == -1 ? arr.length - 1 : index;
    }

    /**
     * 迭代器
     *
     * @return 迭代器
     */
    @Override
    public Iterator<E> iterator() {
        return new Iterator<>() {
            int index = head;
            int cnt = 0;

            @Override
            public boolean hasNext() {
                return cnt < size;
            }

            @Override
            public E next() {
                E value = arr[index];
                index = inc(index);
                cnt++;
                return value;
            }
        };
    }

    /**
     * 重写toString方法
     *
     * @return 字符串
     */
    @Override
    public String toString() {
        StringBuilder builder = new StringBuilder(32);
        builder.append("[");
        int index = head;
        for (int i = 0; i < size; i++) {
            builder.append(arr[index].toString());
            if (index != dec(tail)) {
                builder.append(" ");
            }
            index = inc(index);
        }
        builder.append("]");
        return builder.toString();
    }
}
